package com.kbtd.service.mq;

import cn.hutool.core.bean.BeanUtil;
import com.kbtd.entity.VoucherOrder;
import com.kbtd.service.IVoucherOrderService;
import com.kbtd.service.IVoucherService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * 生成订单消费者线程类
 * Redis实现
 * 缺点：发布消息不可靠，需结合业务操作和lua保证事务一致（业务耦合）
 *
 * @author wp
 */
@Component
@Slf4j
public class VoucherOrderRedisConsumer implements Runnable {
    private static final String queueName = "stream.orders";
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private IVoucherOrderService voucherOrderService;

    @Override
    public void run() {
        while (true) {
            try {
                // 检查 Stream 是否存在
                if (!Boolean.TRUE.equals(stringRedisTemplate.hasKey(queueName))) {
                    // 如果不存在, 重试
                    continue;
                }
                //1.获取redis消息队列中的订单信息 XREADGROUP group g1 c1 count 1 block 2000 stream.orders >
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                        StreamOffset.create(queueName, ReadOffset.lastConsumed())
                );
                //2.判断是否获取成功
                if (list == null || list.isEmpty()) {
                    //获取失败，再来一次
                    continue;
                }
                //3.解析消息的订单信息
                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> values = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                //3.创建订单存到数据库
                voucherOrderService.createVoucherOrder(voucherOrder);
                //4.ACK确认 SACK stream.orders g1 id
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("处理订单异常");
                handlePendingList();
            }
        }
    }

    private void handlePendingList() {
        while (true) {
            try {
                //1.获取pendingList中的订单信息 XREADGROUP group g1 c1 count 1 stream.orders 0
                List<MapRecord<String, Object, Object>> list = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create(queueName, ReadOffset.from("0"))
                );
                //2.判断是否获取成功
                if (list == null || list.isEmpty()) {
                    //获取失败，pendList没有信息结束循环
                    break;
                }
                //3.解析消息的订单信息
                MapRecord<String, Object, Object> record = list.get(0);
                Map<Object, Object> values = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(values, new VoucherOrder(), true);
                //3.创建订单存到数据库
                voucherOrderService.createVoucherOrder(voucherOrder);
                //4.ACK确认 SACK stream.orders g1 id
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", record.getId());
            } catch (Exception e) {
                log.error("处理pendList异常");
                try {
                    Thread.sleep(20);
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        }
    }
}